package chapter4.eg7;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.junit.Test;

public class AvroDeserializerApp {
  @Test
  public void testAvroDeserializer() throws InterruptedException {
    Properties properties = new Properties();
    properties.put("bootstrap.servers", "localhost:9092");
    properties.put("group.id", "CountryCounter2");
    properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    properties.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
    properties.put("schema.registry.url", "http://localhost:8081");

    KafkaConsumer<String, GenericRecord> consumer = new KafkaConsumer<>(properties);
    consumer.subscribe(Collections.singletonList("customerContacts"));
    while (true) {
      ConsumerRecords<String, GenericRecord> records = consumer.poll(Duration.ofMillis(1000));
      for (ConsumerRecord<String, GenericRecord> record : records) {
        GenericRecord customer = record.value();
        System.out.println("Customer id: " + customer.get("id") + ", name: " + customer.get("name"));
      }
      consumer.commitSync();
      TimeUnit.SECONDS.sleep(1);
    }
  }

}
